package com.batyuta.sort;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.log4j.Logger;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;

/**
 *
 */
public class MergeThreadMerge extends AbstractMergeThread<MergeQueueItem> implements Runnable {
    private static Logger logger = Logger.getLogger(MergeThreadMerge.class);

    public MergeThreadMerge(Queue<MergeQueueItem> queueIn, Queue<MergeQueueItem> queueOut, CountDownLatch startSignal, CountDownLatch doneSignal) {
        super(queueIn, queueOut, startSignal, doneSignal);
    }

    @Override
    public MergeQueueItem process() {
        getLogger().info(this.getClass().getName() + "(" + getId() + "): start");
        MergeQueueItem result;
        while (null != (result = poll())) {
            getLogger().info(this.getClass().getName() + "(" + getId() + "): " + result.toString());
            merge(result);
            addToQueue(result);
        }
        if (null != currentQueue) {
            add(currentQueue);
        }
        return result;
    }

    @Override
    Logger getLogger() {
        return logger;
    }
    // TODO: maybe should use Lists for processing but don't read each line from file
    public void merge(MergeQueueItem in) {
        File firstFile = in.getFileFirst(), secondFile = in.getFileSecond(), outputFile = in.getFileOut();
        Integer lineCount = in.getLineCount();
        List<String> result = new LinkedList<String>();
        LineIterator iteratorFirst = null;
        LineIterator iteratorSecond = null;
        OutputStream outputStream = null;
        try {
            outputStream = FileUtils.openOutputStream(outputFile, true);
            iteratorFirst = null != firstFile ? FileUtils.lineIterator(firstFile) : null;
            iteratorSecond = null != secondFile ? FileUtils.lineIterator(secondFile) : null;
            String first = null != iteratorFirst ? getNextString(iteratorFirst) : null;
            String second = null != iteratorSecond ? getNextString(iteratorSecond) : null;
            while (null != first && null != second) {
                if (first.compareTo(second) == 0) {
                    writeToFile(first, result, outputStream, lineCount);
                    writeToFile(second, result, outputStream, lineCount);
                    first = getNextString(iteratorFirst);
                    second = getNextString(iteratorSecond);
                } else if (first.compareTo(second) < 0) {
                    writeToFile(first, result, outputStream, lineCount);
                    first = getNextString(iteratorFirst);
                } else if (first.compareTo(second) > 0) {
                    writeToFile(second, result, outputStream, lineCount);
                    second = getNextString(iteratorSecond);
                }
            }

            while (null != first) {
                writeToFile(first, result, outputStream, lineCount);
                first = getNextString(iteratorFirst);
            }
            while (null != second) {
                writeToFile(second, result, outputStream, lineCount);
                second = getNextString(iteratorSecond);
            }

            if (!result.isEmpty()) {
                FileUtils.writeLines(outputFile, result);
            }
        } catch (IOException e) {
            getLogger().error(e, e);
        } finally {
            IOUtils.closeQuietly(outputStream);
            LineIterator.closeQuietly(iteratorFirst);
            LineIterator.closeQuietly(iteratorSecond);
        }
    }

    public void writeToFile(String first, List<String> result, OutputStream outputStream, Integer lineCount) {
        result.add(first);
        if (result.size() == lineCount) {
            try {
                IOUtils.writeLines(result, null, outputStream, null);
            } catch (IOException e) {
                getLogger().error(e, e);
            }
            result.clear();
        }
    }
}
